package com.ben.utils.flow

import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.SharedFlow
import kotlinx.coroutines.flow.collectLatest
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.first
import java.util.concurrent.atomic.AtomicBoolean

/**
 * Author: bin.yang
 * Maintainer: bin.yang
 * Date: 2021/10/20
 * Copyright: 2021 Inc. All rights reserved.
 * Desc:
 */
class XSharedFlow<T>(private val flow: SharedFlow<T>) {

    private val initStatus = AtomicBoolean(false)

    suspend fun collect(action: suspend (value: T) -> Unit) {
        flow.collect(action)
    }

    suspend fun collectRunOnce(action: suspend (value: T) -> Unit) {
        if (initStatus.compareAndSet(false, true))
            flow.collect(action)
    }

    suspend fun collectLatest(action: suspend (value: T) -> Unit) {
        flow.collectLatest(action)
    }

    suspend fun first(predicate: suspend (T) -> Boolean) {
        flow.first(predicate)
    }

    fun distinctUntilChanged(): XSharedFlow<T> {
        flow.distinctUntilChanged()
        return this
    }

    fun getValue(): T? {
        return flow.replayCache.getOrNull(0)
    }

    fun getFlow(): Flow<T> {
        return flow
    }
}
